package yns.springboot.guide.demo.data.rabbitmq.driver;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Exchange;
import com.rabbitmq.client.AMQP.Queue;

import yns.springboot.guide.demo.configure.data.rabbitmq.RabbitmqConfigure;
import yns.springboot.guide.demo.data.consumer.rabbitmq.driver.RabbitmqDriverConsumer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;

@Configuration
public class RabbitmqDriverFactory {

	// http://192.168.3.14:18000/
	
	@Autowired
	private RabbitmqConfigure rabbitmqConfigure;
	
	@Bean
	public Channel rabbitmqChannelFactory() throws IOException, TimeoutException {
		Channel channel = initChannel();

		String queue = rabbitmqConfigure.getQueue();
		String exchange = rabbitmqConfigure.getExchange();
		String routingKey = rabbitmqConfigure.getRoutingKey();
		
//		channel.exchangeDeclare(exchange, getExchangeType());
		
		channel.queueDeclare(queue, false, false, false, null);
		channel.queueBind(queue, exchange, routingKey);
		
		return channel;
	}
	
	private Channel initChannel() throws IOException, TimeoutException {
		
		ConnectionFactory factory = new ConnectionFactory();

		factory.setHost(rabbitmqConfigure.getHost());
		factory.setPort(Integer.valueOf(rabbitmqConfigure.getPort()));
		factory.setVirtualHost(rabbitmqConfigure.getVirtualHost());

		factory.setUsername(rabbitmqConfigure.getUsername());
		factory.setPassword(rabbitmqConfigure.getPassword());

		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		return channel;
		
	}
	
	private BuiltinExchangeType getExchangeType() {

		String exchangeType = rabbitmqConfigure.getExchangeType();

		if (exchangeType.equalsIgnoreCase(BuiltinExchangeType.DIRECT.getType())) {
			return BuiltinExchangeType.DIRECT;
		} else if (exchangeType.equalsIgnoreCase(BuiltinExchangeType.FANOUT.getType())) {
			return BuiltinExchangeType.FANOUT;
		} else if (exchangeType.equalsIgnoreCase(BuiltinExchangeType.TOPIC.getType())) {
			return BuiltinExchangeType.TOPIC;
		} else if (exchangeType.equalsIgnoreCase(BuiltinExchangeType.HEADERS.getType())) {
			return BuiltinExchangeType.HEADERS;
		}

		return null;
	}
	
	
	
//	public void TestConsumerFactory() throws IOException, TimeoutException {
//		Channel channel = initChannel();
//
//		String queue = rabbitmqConfigure.getQueue();
//		String exchange = rabbitmqConfigure.getExchange();
//		String routingKey = rabbitmqConfigure.getRoutingKey();
//		
//		channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT);
//		
//		channel.queueDeclare(queue, false, false, false, null);
//		channel.queueBind(queue, exchange, routingKey);
//
//		TestConsumer testConsumer = new TestConsumer(channel);
//		channel.basicConsume(queue, true, testConsumer);
//	}
	

	public static void main(String[] args) throws IOException, TimeoutException {
		
		ConnectionFactory factory = new ConnectionFactory();

		factory.setHost("192.168.3.14");
		factory.setPort(5672);
		factory.setVirtualHost("/");

		factory.setUsername("user");// 设置用户名称
		factory.setPassword("123456");// 设置用户密码
		
		String queue = "test";
		String exchange = "exchange.fanout";
		String routingKey = "test.info";
		
		System.out.println("queue : " + queue);
		System.out.println("exchange : " + exchange);
		System.out.println("routingKey : " + routingKey);
		
//		producer(factory,queue,exchange,routingKey);
		consumer(factory,queue,exchange,routingKey);
	}

	public static void producer(ConnectionFactory factory, String queue, String exchange, String routingKey) throws IOException, TimeoutException {

		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		
//		channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT);
		
		channel.queueDeclare(queue, false, false, false, null);
		channel.queueBind(queue, exchange, routingKey);
		
		HashMap<String,Object> headers = new HashMap<>();
		headers.put("main", "producer");
		
		BasicProperties basicProperties = new BasicProperties.Builder()
				.userId("user")
				.headers(headers)
				.contentType("text/plan")
				.priority(1)
				.build();
		
		String message = LocalDateTime.now().toString();
		channel.basicPublish(exchange, routingKey, basicProperties, message.getBytes("UTF-8"));
		System.out.println(" [Producer] Sent  : " + message);

		channel.close();
		connection.close();
	}

	public static void consumer(ConnectionFactory factory, String queue, String exchange, String routingKey) throws IOException, TimeoutException {

		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
//		channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT);
		
		channel.queueDeclare(queue, false, false, false, null);
		channel.queueBind(queue, exchange, routingKey);

		Queue.DeclareOk declareOk = channel.queueDeclarePassive(queue);
		System.out.println("Queue : " + declareOk.getQueue());
		System.out.println("MessageCount : " + declareOk.getMessageCount());
		System.out.println("ConsumerCount : " + declareOk.getConsumerCount());
		
		RabbitmqDriverConsumer testConsumer = new RabbitmqDriverConsumer(channel);
		channel.basicConsume(queue, true, testConsumer);
		

//		GetResponse getResponse  = channel.basicGet(queue, false);
//		String body = new String(getResponse.getBody(), "UTF-8");
//		System.out.println(" [Consumer] Received  : " + body );
//		System.out.println(" Envelope :   " + getResponse.getEnvelope());
//		System.out.println(" BasicProperties :   " + getResponse.getProps());
		
//		channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false); // 消息确认


//		channel.close();
//		connection.close();
	}

}
